﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net;
using System.Net.Sockets;
using System.Threading.Tasks;

using WorkStation.Frameworks.Core;
using WorkStation.Frameworks.Core.Extensions;

using WorkStation.Frameworks.EasyBus.Extensions;
using WorkStation.Frameworks.EasyBus.Serializable;
using WorkStation.Frameworks.EasyBus.Plugins;
using WorkStation.Frameworks.EasyBus.DbModel;
using WorkStation.Frameworks.EasyBus.Handles;
using WorkStation.Frameworks.EasyBus.Repositories;
using WorkStation.Frameworks.EasyBus.Attributes;
using WorkStation.Frameworks.EasyBus.MessageQueue;
using WorkStation.Frameworks.EasyBus.Pipeline;
using WorkStation.Frameworks.EasyBus.Pipeline.Extensions;
using WorkStation.Frameworks.EasyBus.Pipeline.Handles;
using WorkStation.Frameworks.EasyBus.Pipeline.Filters;

using Newtonsoft.Json;
using Microsoft.Practices.ServiceLocation;

namespace WorkStation.Frameworks.EasyBus.Extensions
{
    /// <summary>
    /// 总线对象扩展
    /// </summary>
    public static class IBusExtensions
    {
        /// <summary>
        /// 设置插件
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <param name="plugin">插件对象</param>
        /// <returns></returns>
        public static IBus SetPlugin(this IBus bus, IBusPlugin plugin) 
        {
            return bus.Continue(v => bus.Plugins.Add(plugin));
        }

        /// <summary>
        /// 获取插件
        /// </summary>
        /// <typeparam name="TPlugin">插件类型</typeparam>
        /// <param name="bus">总线对象</param>
        /// <returns></returns>
        public static TPlugin GetPlugin<TPlugin>(this IBus bus) where TPlugin : IBusPlugin
        {
            return (TPlugin)bus.Plugins.FirstOrDefault(v => v is TPlugin);
        }

        /// <summary>
        /// 设置默认的错误日志
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <returns></returns>
        public static IBus SetDefaultErrorLogger(this IBus bus)
        {
            bus.OnEventFail += (context, result) => bus.Logger.Error(result.Error);
            bus.OnPublishFail += (context, result) => bus.Logger.Error(result.Error);
            bus.OnProcessFail += (context, result) => bus.Logger.Error(result.Error);
            bus.OnListenException += ce => bus.Logger.Error(ce);

            return bus;
        }

        /// <summary>
        /// 设置总线的消息注册中心仓储组件
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <param name="repository">仓储对象</param>
        /// <returns></returns>
        public static IBus SetRegisterCenterRepository(this IBus bus, IRegisterCenterRepository repository) 
        {
            IBusDataManager manager = bus as IBusDataManager;

            manager.RegisterCenterRepository = repository;

            bus.UploadPublishInfo()
               .UploadSubscribeInfo()
               .UploadListenerInfo()
               .InitPublishByRegisterCenter()
               .InitListenerByRegisterCenter()
               .Subscribe<object>(new MessageRouteHandle().Init(bus));

            return bus;
        }

        /// <summary>
        /// 设置总线的历史记录中心仓储组件
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <param name="repository">仓储对象</param>
        /// <returns></returns>
        public static IBus SetHistoryCenterRepository(this IBus bus, IHistoryCenterRepository repository) 
        {
            IBusDataManager manager = bus as IBusDataManager;

            manager.HistoryCenterRepository = repository;

            SetupMsgLog(bus);

            return bus;
        }

        /// <summary>
        /// 设置配置提供器组件
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <param name="provider">提供器</param>
        /// <returns></returns>
        public static IBus SetConfigProvider(this IBus bus, IConfigProvider provider) 
        {
            return bus.Continue(v => v.ConfigProvider.Add(provider));
        }

        /// <summary>
        /// 设置标记配置提供器
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <param name="assemblyNames">消息所在程序集</param>
        /// <returns></returns>
        public static IBus SetAttributeConfigProvider(this IBus bus, params string[] assemblyNames) 
        {
            return bus.Continue(v => v.ConfigProvider.Add(new AttributeConfigProvider(assemblyNames) { Bus = bus }));
        }

        /// <summary>
        /// 订阅发布处理
        /// </summary>
        /// <typeparam name="TMsg">消息类型（将消息类型作为路由关键字）</typeparam>
        /// <param name="bus">总线对象</param>
        /// <param name="handle">处理器</param>
        /// <returns></returns>
        private static IBus Subscribe<TMsg>(this IBus bus, IHandle handle) 
        {
            IBusDataManager manager = bus as IBusDataManager;

            manager.SendMessagePipeline.Subscribe<TMsg>(handle);

            return bus;
        }

        /// <summary>
        /// 订阅发布处理
        /// </summary>
        /// <typeparam name="TMsg">消息类型（将消息类型作为路由关键字）</typeparam>
        /// <param name="bus">总线对象</param>
        /// <param name="handle">处理委托</param>
        /// <param name="handleName">处理器名称</param>
        /// <param name="filter">过滤器</param>
        /// <param name="order">排序</param>
        /// <returns></returns>
        public static IBus Subscribe<TMsg>(this IBus bus, Func<ActionContext, IHandle, HandleResult> handle, string handleName = null, IFilter filter = null, int order = -1) 
        {
            IBusDataManager manager = bus as IBusDataManager;

            manager.SendMessagePipeline.Subscribe<TMsg>(handle, handleName, filter, order);

            return bus;
        }

        /// <summary>
        /// 按照订阅信息订阅发布
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <param name="info">订阅信息</param>
        /// <returns></returns>
        public static IBus Subscribe(this IBus bus, SubscribeInfo info) 
        {
            IEndpoint endpoint = info.Mq.ToEndpoint();

            if (endpoint == null)
            {
                EasyBusException e = new EasyBusException("订阅端终结点地址翻译异常");

                e.Data["SubscribeInfo"] = info;

                throw e;
            }

            bus.Subscribe(info.RouteKey, new DefaultPublishHandle() { TargetEndpoint = endpoint, Name = DefaultPublishHandle.GetHandleName(info), SubscribeBusName = info.BusName, SubscribeServerNode = info.NodeName, SubscribeName = info.Name });

            return bus;
        }

        /// <summary>
        /// 按照监听信息设置监听
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <param name="info">监听信息</param>
        /// <returns></returns>
        public static IBus Listen(this IBus bus, ListenerInfo info) 
        {
            IEndpoint endpoint = info.Mq.ToEndpoint();

            bus.Listen(info.Name, endpoint, info.ProcessThreadCount);

            foreach (string handleName in info.Process.HandleName) 
            {
                ProcessHandle handle = null;

                try
                {
                    handle = ServiceLocator.Current.GetInstance<ProcessHandle>(handleName);
                }
                catch (Exception ce) 
                {
                    throw new EasyBusException("获取处理器异常，HandleName：{0}，详细见内部异常".Fmt(handleName), ce);
                }

                bus.SetListenerProcess(info.Name, info.Process.RouteKey, handle);
            }

            return bus;
        }

        /// <summary>
        /// 初始化发布根据注册中心
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <returns></returns>
        public static IBus InitPublishByRegisterCenter(this IBus bus) 
        {
            IBusDataManager manager = bus as IBusDataManager;

            try
            {
                List<SubscribeInfo> info = manager.RegisterCenterRepository.GetBusSubscribeInfo(bus.Name);

                bus.DataBag["SubscribeInfo"] = info;

                // Register handle 
                info.ForEach(v => bus.Subscribe(v));
            }
            catch (EasyBusException ce) 
            {
                throw ce;
            }
            catch (Exception ce)
            {
                throw new EasyBusException("使用注册中心初始化发布异常，详情见内部异常", ce);
            }

            return bus;
        }

        /// <summary>
        /// 初始化监听根据注册中心
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <returns></returns>
        public static IBus InitListenerByRegisterCenter(this IBus bus) 
        {
            IBusDataManager manager = bus as IBusDataManager;

            try
            {
                List<ListenerInfo> info = manager.RegisterCenterRepository.GetBusListenerInfoByNode(bus.Name, bus.NodeName);

                info.ForEach(v => bus.Listen(v));
            }
            catch (Exception ce) 
            {
                throw new EasyBusException("使用注册中心初始化监听器异常，详情见内部异常", ce);
            }

            return bus;
        }

        /// <summary>
        /// 设置消息日志
        /// </summary>
        /// <param name="bus">消息总线对象</param>
        /// <returns></returns>
        private static void SetupMsgLog(IBus bus)
        {
            IBusDataManager manager = bus as IBusDataManager;

            bus.OnPublishBefore += context =>
            {
                if ((context.Input as Message).IsRecordingHistory)
                {
                    Task.Factory.StartNew(state =>
                    {
                        BusActionContext innerContext = state as BusActionContext;

                        Message msg = innerContext.Input as Message;

                        MessageLog log = new MessageLog()
                        {
                            MessageId = msg.MessageId,
                            RouteKey = innerContext.RouteKey,
                            Json = msg.BodyJson,
                            PublishServerIp = manager.Ip,
                            PublishBusName = bus.Name,
                            PublishBusNodeName = bus.NodeName,
                            CreateTime = DateTime.Now
                        };

                        try
                        {
                            manager.HistoryCenterRepository.UploadMessageLog(log);
                        }
                        catch (Exception ce)
                        {
                            EasyBusException e = new EasyBusException("上传消息日志异常，详细见内部异常", ce);

                            bus.Logger.Error(e.ToJsonString());

                            throw e;
                        }
                    }, context);
                }
            };

            bus.OnPublishAfter += (context, results) =>
            {
                Message msg = context.Input as Message;

                // 上传消息日志的发布标志位
                if (msg.IsRecordingHistory)
                {
                    bool isSuccessPublish = true;

                    // 如果结果不是全发布成功则上传标志位为 false 
                    if (!results.Select(v => v as PublishHandleResult).Where(v => v != null).All(v => (v as PublishHandleResult).PublishResult.FirstOrDefault().IsSuccess))
                    {
                        isSuccessPublish = false;   
                    }

                    Task.Factory.StartNew(() =>
                    {
                        try
                        {
                            manager.HistoryCenterRepository.UpdateIsSuccessPublish(msg.MessageId, isSuccessPublish);
                        }
                        catch (Exception ce)
                        {
                            EasyBusException e = new EasyBusException("上传消息日志的发布标志位异常，详细见内部异常", ce);

                            bus.Logger.Error(e.ToJsonString());

                            throw e;
                        }
                    });
                }

                foreach (HandleResult result in results)
                {
                    // 上传发布日志到消息历史记录中心
                    if (result is PublishHandleResult && msg.IsRecordingHistory)
                    {
                        Task.Factory.StartNew(r =>
                        {
                            PublishHandleResult publishHandleResult = r as PublishHandleResult;

                            try
                            {
                                foreach (MessagePublishLog log in publishHandleResult.PublishResult)
                                {
                                    manager.HistoryCenterRepository.UploadMessagePublishLog(log);
                                }
                            }
                            catch (Exception ce)
                            {
                                EasyBusException e = new EasyBusException("上传发布日志异常，详细见内部异常", ce);

                                bus.Logger.Error(e.ToJsonString());

                                throw e;
                            }
                        }, result);
                    }

                    // 总线路由处理器发生异常
                    if (result is MessageRouteHandleErrorResult)
                    {
                        EasyBusException e = new EasyBusException("总线路由处理器【{0}】发生异常".Fmt(typeof(MessageRouteHandle).FullName), (result as FailHandleResult).Error);

                        bus.Logger.Error(e.ToJsonString());

                        throw e;
                    }
                }
            };

            bus.OnProcessAfter += (context, results) => 
            {
                Message msg = context.Input as Message;

                // 上传消息处理标志位信息
                if (msg.IsRecordingHistory) 
                {
                    bool isSuccessProcess = true;

                    if (results.Any(v => v is IProcessHandleResult && v is ProcessFailResult)) 
                    {
                        isSuccessProcess = false;
                    }

                    Task.Factory.StartNew(() => 
                    {
                        try
                        {
                            manager.HistoryCenterRepository.UpdateIsSuccessProcess(msg.MessageId, isSuccessProcess);
                        }
                        catch (Exception ce) 
                        {
                            EasyBusException e = new EasyBusException("上传消息日志的处理标志位异常，详细见内部异常", ce);

                            bus.Logger.Error(e.ToJsonString());

                            throw e;
                        }
                    });
                }

                foreach (HandleResult result in results)
                {
                    // 总线监听路由处理器发生异常
                    if (result is ProcessRouteHandleErrorResult)
                    {
                        EasyBusException e = new EasyBusException("总线监听路由处理器【{0}】发生异常".Fmt(typeof(ProcessRouteHandle).FullName), (result as ProcessFailResult).Error);

                        bus.Logger.Error(e.ToJsonString());

                        throw e;
                    }

                    // 上传处理日志
                    if (result is IProcessHandleResult && (context.Input as Message).IsRecordingHistory)
                    {
                        Task.Factory.StartNew(r =>
                        {
                            HandleResult innerResult = r as HandleResult;

                            MessageProcessLog log = new MessageProcessLog()
                            {
                                MessageId = msg.MessageId,
                                HandleName = innerResult.Handle.Name,
                                Mq = (innerResult as IProcessHandleResult).Endpoint.ToEndpointString(),
                                BusName = bus.Name,
                                NodeName = bus.NodeName,
                                ListenerName = (innerResult.Handle as ProcessHandle).Listener.Name
                            };

                            if (innerResult is ProcessFailResult)
                            {
                                log.IsSuccess = false;
                                log.Error = (innerResult as ProcessFailResult).ErrorMessage;
                            }
                            else
                            {
                                log.IsSuccess = true;
                            }

                            try
                            {
                                manager.HistoryCenterRepository.UploadMessageProcessLog(log);
                            }
                            catch (Exception ce)
                            {
                                EasyBusException e = new EasyBusException("上传处理日志失败", ce);

                                bus.Logger.Error(e.ToJsonString());

                                throw e;
                            }
                        }, result);
                    }
                }
            };
        }

        /// <summary>
        /// 上传总线的发布信息
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <returns></returns>
        private static IBus UploadPublishInfo(this IBus bus)
        {
            IBusDataManager manager = bus as IBusDataManager;

            bus.ConfigProvider.ForEach(v =>
            {
                manager.RegisterCenterRepository.UploadPublishInfo(v.GetPublishInfo().ToList());
            });

            return bus;
        }

        /// <summary>
        /// 上传总线的订阅信息
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <returns></returns>
        private static IBus UploadSubscribeInfo(this IBus bus)
        {
            IBusDataManager manager = bus as IBusDataManager;

            bus.ConfigProvider.ForEach(v =>
            {
                manager.RegisterCenterRepository.UploadSubscribeInfo(v.GetSubscribeInfo().ToList());
            });

            return bus;
        }

        /// <summary>
        /// 上传总线的监听信息
        /// </summary>
        /// <param name="bus">总线对象</param>
        /// <returns></returns>
        private static IBus UploadListenerInfo(this IBus bus)
        {
            IBusDataManager manager = bus as IBusDataManager;

            bus.ConfigProvider.ForEach(v =>
            {
                manager.RegisterCenterRepository.UploadListenerInfo(v.GetListenerInfo().ToList());
            });

            return bus;
        }
    }
}
